package com.atguigu.sparkstreaming.output

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 *
 *  windows版本无法运行shell命令！
 *
 *
 *  C:\MyFile\Dev\hadoop-3.1.0\bin\winutils.exe
 *      chmod 0644
 *   E:\ideaproject\project220309\wordcount-1657869780000..txt\_temporary\0\_temporary\attempt_20220715152300_0004_m_000000_0\part-00000
 *
 */
object SaveAsTextFileDemo {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "TransformDemo", Seconds(5))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "220309",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "true"
    )


    val topics = Array("topicA")

    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )
    

    //(String, Int):(word,n)
    val ds2: DStream[(String, Int)] = ds.flatMap(record => record.value().split(" "))
      .map((_, 1)).reduceByKey(_+_)

    //以文件形式输出
    ds2.saveAsTextFiles("wordcount",".txt")

    // 启动APP
    streamingContext.start()

    // 阻塞进程，让进程一直运行
    streamingContext.awaitTermination()

  }

}
